package cn.jhc.teachingquality.rx;

import rx.Observable;
import rx.functions.Func1;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;

/**
 * RxBus事件总线实现，参考自RxVolley。
 * Created by luyanfei on 16-6-2.
 */
public class RxBus {
    private static volatile RxBus instance;
    private final SerializedSubject<Object, Object> subject =
            new SerializedSubject<>(PublishSubject.create());

    public static RxBus getInstance() {
        if (instance == null) {
            synchronized (RxBus.class) {
                if (instance == null) {
                    instance = new RxBus();
                }
            }
        }
        return instance;
    }

    public void post(Object event) {
        subject.onNext(event);
    }

    public <T> Observable<T> take(final Class<T> eventType) {
        return subject.filter(new Func1<Object, Boolean>() {
            @Override
            public Boolean call(Object o) {
                return eventType.isInstance(o);
            }
        }).cast(eventType);
    }
}
